package com.mercury.kafka.manager;

import com.mercury.kafka.domain.KafKaMqMessage;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * @author mercury
 */
@Component
@Slf4j
public class KafkaManager {


    @Resource
    private KafkaTemplate<String, KafKaMqMessage> kafkaTemplate;

    /**
     * 发送kafka消息
     *
     * @param topic   主题
     * @param message 消息内容
     */
    public void sendMsg(String topic, KafKaMqMessage message) {
        Assert.notNull(message, "message is null");
        if (topic == null || "".equals(topic.trim())) {
            return;
        }
        //声明kafka事物
        kafkaTemplate.executeInTransaction(operations -> {
            ProducerRecord<String, KafKaMqMessage> consumerRecord = new ProducerRecord<>(topic, message);
            //发消息
            ListenableFuture<SendResult<String, KafKaMqMessage>> future = operations.send(consumerRecord);
            future.addCallback(new ListenableFutureCallback<SendResult<String, KafKaMqMessage>>() {
                @Override
                public void onSuccess(SendResult<String, KafKaMqMessage> result) {
                    log.info(("Sent message=[" + message.getId() + "] with offset=[" + result.getRecordMetadata().offset() + "]"));
                }

                @Override
                public void onFailure(@NonNull Throwable ex) {
                    log.error("Unable to send message=[" + message.getId() + "] due to : " + ex.getMessage());
                }
            });
            return future;
        });
    }
}
